package org.promote.hotspot.client.etcd;

import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Watch;
import io.etcd.jetcd.watch.WatchEvent;
import io.grpc.Server;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.promote.hotspot.client.ClientContext;
import org.promote.hotspot.client.core.server.ServerInfoChangeEvent;
import org.promote.hotspot.client.eventbus.EventBusCenter;
import org.promote.hotspot.client.callback.ReceiveNewKeyEvent;
import org.promote.hotspot.client.rule.RuleChangeEvent;
import org.promote.hotspot.common.config.ConfigConstant;
import org.promote.hotspot.common.etcd.JetcdClient;
import org.promote.hotspot.common.model.HotKeyModel;
import org.promote.hotspot.common.tool.FastJsonUtils;
import org.promote.hotspot.common.rule.KeyRule;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static java.nio.charset.StandardCharsets.UTF_8;

/**
 * etcd启动类
 *
 * @author enping.jep
 * @date 2023/10/24 15:29
 **/
@Slf4j
public class EtcdLauncher {

    public void launch() {
        connectServer(); //定时拉取服务端信息并连接
        initRule();//初始化规则信息、初始化热点数据的缓存信息
        startWatchRule();//监听规则变化
    }


    /**
     * 拉取规则信息
     */
    private void initRule() {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        //开启拉取etcd的规则信息，如果拉取失败，则定时继续拉取
        scheduledExecutorService.scheduleAtFixedRate(() -> {
//            log.info(getClass() + "：trying to connect to etcd and fetch rule info");
            boolean success = fetchRuleFromEtcd();//保障重要数据的高可用： 冗余  重试
            if (success) {
                //如果成功获取，则关闭线程池，如果获取失败，则每5秒获取一次
                scheduledExecutorService.shutdown();
            }
        }, 0, 5, TimeUnit.SECONDS);
    }

    /**
     * 异步监听rule规则变化
     */
    private void startWatchRule() {
        JetcdClient jetcdClient = EtcdConfigFactory.getJetcdClient();
        String key = ConfigConstant.RULE_PATH + ClientContext.APP_NAME;
        log.info(getClass() + "--- begin watch rule change ----");
        jetcdClient.watch(key, Watch.listener(watchResponse -> {
            log.info("收到[{}]的事件", key);
            // 被调用时传入的是事件集合，这里遍历每个事件
            watchResponse.getEvents().forEach(watchEvent -> {
                // 操作类型
                WatchEvent.EventType eventType = watchEvent.getEventType();
                // 操作的键值对
                KeyValue keyValue = watchEvent.getKeyValue();
                log.info("type={}, key={}, value={}",
                        eventType,
                        keyValue.getKey().toString(UTF_8),
                        keyValue.getValue().toString(UTF_8));
                String rules = keyValue.getValue().toString(UTF_8);
                notifyRuleChange(rules);
            });
        }));
    }

    /**
     * 初始化以及更新规则时使用
     *
     * @param rules 规则字符串
     */


    /**
     * 从etcd获取规则
     * 如果获取成功在返回true，失败返回false
     * 获取最新规则
     * 更新本地的缓存及规则列表
     */
    private boolean fetchRuleFromEtcd() {
        JetcdClient jetcdClient = EtcdConfigFactory.getJetcdClient();
        try {
            //从etcd获取自己的rule
            String rules = jetcdClient.get(ConfigConstant.RULE_PATH + ClientContext.APP_NAME);
            notifyRuleChange(rules);
            return true;
        } catch (Exception e) {
            log.error(getClass() + ":fetch rule failure, please check the rule info in etcd");
            return false;
        }
    }

    private void notifyRuleChange(String rules) {
        EventBusCenter.getInstance().post(new RuleChangeEvent(rules));//通知机制,向订阅者发送消息
    }

    /**
     * 定时拉取服务端信息
     * 不会被频繁调用，只会在启动的时候，调用一次
     */
    private void connectServer() {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        //开启拉取etcd的server信息，如果拉取失败，则定时继续拉取
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            log.info(getClass() + ":trying to connect to etcd and fetch server info");
            fetchServer();
        }, 0, 30, TimeUnit.SECONDS);
    }

    private void fetchServer() {
        JetcdClient jetcdClient = EtcdConfigFactory.getJetcdClient();
        //获取所有server的地址  先从/hotspot/servers/$APP_NAME中获取，允许该APP有属于自己的server集群
        //app热点特别多，需要有自己专有的server
        List<KeyValue> keyValues = jetcdClient.getPrefix(ConfigConstant.SERVERS_PATH + ClientContext.APP_NAME);
        //如果没有，则使用默认的server
        if (CollectionUtils.isEmpty(keyValues)) {
            //用公共的server
            keyValues = jetcdClient.getPrefix(ConfigConstant.SERVERS_PATH + "default");
        }

        if (CollectionUtils.isEmpty(keyValues)) {
            log.warn(getClass() + "server ip is null!");
            return;
        }

        List<String> addresses = new ArrayList<>();
        for (KeyValue keyValue : keyValues) {
            //value里放的是ip地址
            String ipPort = keyValue.getValue().toString(StandardCharsets.UTF_8);
            log.info("获取服务端地址信息：" + ipPort);
            addresses.add(ipPort);
        }

        //发布服务端的变更信息
        notifyServerInfoChange(addresses);
    }

    private void notifyServerInfoChange(List<String> addresses) {
        EventBusCenter.getInstance().post(new ServerInfoChangeEvent(addresses));
    }
}
